package io.gitee.icefairy.swclient4flink;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.FileSystemFactory;
import seaweedfs.client.FilerClient;

import java.io.IOException;
import java.net.URI;

public class SwFileSystemFactory implements FileSystemFactory {
    private Configuration config = new Configuration();

    @Override
    public String getScheme() {
        return "sw";
    }

    @Override
    public void configure(Configuration config) {
        if (null == config) {
            throw new RuntimeException("config cannot be empty");
        } else {
            config.keySet().forEach(s -> {
                String k = s;
                if (k.startsWith("fs.sw.")) {
                    String newKey = k.substring(6);
                    config.setString(newKey, config.getString(k, ""));
                }
            });
        }
    }

    @Override
    public FileSystem create(URI fsUri) throws IOException {
        String host = fsUri.getHost() == null ? config.getString("filer_host", "localhost") : fsUri.getHost();
        int port = fsUri.getPort() == 0 ? config.getInteger("filer_grpc_port", 18888) : fsUri.getPort();
        FilerClient fc = new FilerClient(host, port);
        SwFileSystem swfs = new SwFileSystem(fc);
        return swfs;
    }
}
